/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.streaming;

import org.apache.spark.api.java.JavaRDD;

import java.util.Iterator;

/**
 * SparkStreamPartitionService
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
public interface SparkStreamPartitionService<T> extends SparkStreamRddService<T> {
    /**
     * 处理每个RDD
     *
     * @param rdd JavaRDD<T>
     */
    public default void doEachRDD(final JavaRDD<T> rdd) {
        rdd.foreachPartition(partition -> {
            if (partition.hasNext()) {
                try {
                    initEachPartition(partition);
                } catch (Exception e) {
                    e.printStackTrace();
                    //partition初始失败就直接结束程序
                    System.exit(0);
                }
                try {
                    doEachPartition(partition);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    postEachPartition(partition);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }


    /**
     * 对每个分区进行初始化操作，比如获取数据库连接，网络连接等。
     *
     * @param partition Iterator<T>
     */
    public void initEachPartition(final Iterator<T> partition);

    /**
     * 对每个分区进行业务处理
     * <li>此处是最原始的数据未进行过封装，不建议重写此方法</li>
     *
     * @param partition Iterator<T>
     */
    public void doEachPartition(final Iterator<T> partition);

    /**
     * 每个Partition处理完成后的回调,比如释放数据库连接，网络连接等
     *
     * @param partition Iterator<T>
     */
    public void postEachPartition(final Iterator<T> partition);

}

